package com.zhouyu.fileSync.FSclient.transmission.communicationBetweenClients.client;

import com.zhouyu.fileSync.FSclient.ClientAppContext;
import com.zhouyu.fileSync.FSclient.InvocationHandlers.RpcInvocationHandler;
import com.zhouyu.fileSync.FSclient.transmission.communicationBetweenClients.api.CommunicationService;
import com.zhouyu.fileSync.FSclient.transmission.communicationBetweenClients.api.impl.CommunicationServiceImpl;
import com.zhouyu.fileSync.FScommon.api.FileAndJsonDecoder;
import com.zhouyu.fileSync.FScommon.api.JSONEncoder;
import com.zhouyu.fileSync.FScommon.api.RpcNettyClient;
import com.zhouyu.fileSync.FScommon.api.entities.Request;
import com.zhouyu.fileSync.FScommon.api.entities.Response;
import com.zhouyu.fileSync.FScommon.entities.FSClient;
import io.netty.bootstrap.Bootstrap;
import io.netty.buffer.Unpooled;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.codec.DelimiterBasedFrameDecoder;
import io.netty.handler.timeout.IdleStateHandler;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;

import java.util.concurrent.SynchronousQueue;
import java.util.concurrent.TimeUnit;

import static com.zhouyu.fileSync.FSclient.ClientAppContext.REGISTER_SERVICE_RPC;

public class CommunicationClient implements RpcNettyClient {

    Logger logger = LogManager.getLogger(CommunicationClient.class);


    private CommunicationClientHandler CommunicationClientHandler = new CommunicationClientHandler();

    private String ccId;

    private Channel channel;

    private FSClient initiator;

    private FSClient target;

    private CommunicationService communicationService;


    public CommunicationClient(FSClient initiator, FSClient target) {
        this.initiator = initiator;
        if (ClientAppContext.TARGETS.containsKey(target.getClientName())){
            this.target = ClientAppContext.TARGETS.get(target.getClientName());
        }else {
            while (true){
                FSClient fsc=REGISTER_SERVICE_RPC.getClient(target.getClientName(),target.getClientPassword());
                if (fsc!=null) {
                    ClientAppContext.TARGETS.put(target.getClientName(), fsc);
                    this.target=fsc;
                    break;
                }else {
                    //每隔一分钟重试一次
                    try {
                        TimeUnit.MINUTES.sleep(1);
                        logger.info("try to connect to client :"+target.getClientName());
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                        logger.error(e);
                    }

                }
            }
        }
        this.communicationService = (CommunicationService) new RpcInvocationHandler().newProxyInstance(new CommunicationServiceImpl(), this);
        this.ccId=initiator.getClientHost()+initiator.getClientHost()+target.getClientHost()+target.getClientHost();
    }

    @Override
    public void run() {
        EventLoopGroup group = new NioEventLoopGroup();
        try {
            Bootstrap bootstrap = new Bootstrap();
            bootstrap.group(group).channel(NioSocketChannel.class)
                    .option(ChannelOption.SO_KEEPALIVE, true)
                    .option(ChannelOption.TCP_NODELAY, true).handler(new ChannelInitializer<SocketChannel>() {

                protected void initChannel(SocketChannel socketChannel) throws Exception {
                    ChannelPipeline pipeline = socketChannel.pipeline();
                    pipeline.addLast(new IdleStateHandler(0, 0, 10));
                    pipeline.addLast(new DelimiterBasedFrameDecoder(65535, Unpooled.copiedBuffer("#_@_#".getBytes())));
                    pipeline.addLast(new JSONEncoder());
                    pipeline.addLast(new FileAndJsonDecoder(ClientAppContext.CONFIGS.CLIENT.getTempPath()));
                    pipeline.addLast(CommunicationClientHandler);
                }
            });

            ChannelFuture cf = bootstrap.connect(target.getClientHost(), target.getClientPort()).sync();
            logger.info("file node client start up! connect to host:" + target.getClientHost() + ":" + target.getClientPort());
            //等待服务端监听端口关闭
            channel = cf.channel();
            channel.closeFuture().sync();
        } catch (Exception e) {
            e.printStackTrace();
            logger.error(e);
        } finally {
            group.shutdownGracefully();
        }
    }

    @Override
    public Response sendToServer(Request request) throws InterruptedException {
        if (channel != null && channel.isActive()) {
            SynchronousQueue<Object> queue = CommunicationClientHandler.sendRequest(request, channel);
            Object response = queue.take();
            return (Response) response;
        } else {
            Response response = new Response();
            response.setCode(1);
            response.setErrorMsg("failed to connect to server,please check the configs!");
            return response;
        }

    }

    public Channel getChannel() {
        return channel;
    }


    public FSClient getInitiator() {
        return initiator;
    }

    public void setInitiator(FSClient initiator) {
        this.initiator = initiator;
    }

    public FSClient getTarget() {
        return target;
    }

    public void setTarget(FSClient target) {
        this.target = target;
    }

    public CommunicationService getCommunicationService() {
        return communicationService;
    }

    public String getCcId() {
        return ccId;
    }

    public void setCcId(String ccId) {
        this.ccId = ccId;
    }
}
